package com.merry.kafka.test.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * @author zouyb
 * 自定义kafka拦截器实现相关接口 ProducerInterceptor
 *
 */
public class ValueInterceptor implements ProducerInterceptor<String,String> {

    @Override
    //发送数据时，调用该方法
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        //拦截相关的数据。
        return new ProducerRecord<String,String>
                (record.topic(),record.key(),record.value()+record.value());
    }

    @Override
    //发送数据完毕，服务器返回的响应，会调用该方法
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {

    }

    @Override
    //生产者对象关闭的时候，调用该方法
    public void close() {

    }

    @Override
    //创建生产者对象的时候，调用该方法。可以在这里面执行一些登录验证操作。
    public void configure(Map<String, ?> configs) {
        //从这里面获取的配置信息，一定要在创建生产者对象的时候指定。
        Object password = configs.get("password");
        Object username = configs.get("username");
        //登录一些第三方的组件。
    }
}
